package com.bw.medical.realtime.app.dwd;

import com.bw.medical.realtime.common.MedicalCommon;
import com.bw.medical.realtime.util.CreateEnvUtil;
import com.bw.medical.realtime.util.KafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/*
9.7 医生域医生注册事务事实表
 */

public class DwdDoctorRegister {
    public static void main(String[] args) {
        // TODO 1. 初始化表处理环境
        StreamExecutionEnvironment env = CreateEnvUtil.getStreamEnv(8087, "medical_dwd_trade_consultation");
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        env.setParallelism(1);

        // TODO 2. 读取数据
        String topic = MedicalCommon.KAFKA_ODS_TOPIC;
        String groupId = "medical_dwd_doctor_register";
        CreateEnvUtil.createOdsTable(tableEnv, topic, groupId);

        // TODO 3. 筛选数据
        Table resultTable = tableEnv.sqlQuery("select\n" +
                "`data`['id'] id,\n" +
                "`data`['create_time'] register_time,\n" +
                "`data`['birthday'] birthday,\n" +
                "`data`['consultation_fee'] consultation_fee,\n" +
                "`data`['gender'] gender_code,\n" +
                "`data`['specialty'] specialty_code,\n" +
                "`data`['title'] title_code,\n" +
                "`data`['hospital_id'] hospital_id\n" +
                "from `topic_db`\n" +
                "where `table` = 'doctor'\n" +
                "and `type` = 'insert'");
        tableEnv.createTemporaryView("result_table", resultTable);

        // TODO 4. 写入 Kafka 目标主题
        String sinkTopic = "medical_dwd_doctor_register";
        tableEnv.executeSql("create table " + sinkTopic + "(\n" +
                "`id` string,\n" +
                "`create_time` string,\n" +
                "`birthday` string,\n" +
                "`consultation_fee` string,\n" +
                "`gender` string,\n" +
                "`specialty` string,\n" +
                "`title` string,\n" +
                "`hospital_id` string\n" +
                ") " + KafkaUtil.getKafkaSinkDDL(sinkTopic));
        tableEnv.executeSql("insert into " + sinkTopic + " select * from result_table");
    }
}
